package com.emqx.controller;

import com.emqx.enums.QosEnum;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.util.ObjectUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;

@RestController
@RequestMapping("/mutil")
public class MutilPublishController {


    private Map<String, MqttClient> mqttClientMap = new HashMap<>();
    private Map<String, Integer> clientStatusMap = new HashMap<>();

    private static final Integer RUNNING = 1;
    private static final Integer SLEEPING = 0;

    @GetMapping("/connect")
    public void connect(@RequestParam(value = "broker") String broker,
                        @RequestParam(value = "username") String username,
                        @RequestParam(value = "password") String password,
                        @RequestParam(value = "clientId") String clientId){
        try {
            MqttClient client = new MqttClient(broker, clientId, new MemoryPersistence());

            // 连接参数
            MqttConnectOptions options = new MqttConnectOptions();

            // 设置用户名和密码
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setConnectionTimeout(60);
            options.setKeepAliveInterval(60);
            // 连接
            client.connect(options);
            System.out.println("对接mqtt服务器成功");
            System.out.println("broker: " + broker);
            System.out.println("username: " + username);
            System.out.println("clientId: " + clientId);

            // 存放连接元素
            mqttClientMap.put(broker, client);
            clientStatusMap.put(broker, RUNNING);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    @GetMapping("/publish")
    public void publish(@RequestParam(value = "broker") String broker,
                        @RequestParam(value = "topic") String topic,
                        @RequestParam(value = "content") String content,
                        @RequestParam(value = "qos") Integer qos){
        MqttClient client = mqttClientMap.get(broker);
        if(ObjectUtils.isEmpty(client)){
            System.out.println("mqtt服务器尚未连接");
        }

        Integer status = clientStatusMap.get(broker);
        if(SLEEPING.equals(status)){
            try {
                System.out.println("mqtt服务器正在睡眠...");
                System.out.println("尝试唤醒mqtt服务器" + broker);
                client.connect();
                System.out.println("成功唤醒mqtt服务器" + broker);
            } catch (MqttException e) {
                throw new RuntimeException(e);
            }
            clientStatusMap.put(broker, RUNNING);
        }

        // 创建消息并设置 QoS
        MqttMessage message = new MqttMessage(content.getBytes());
        message.setQos(qos);

        // 发布消息
        try {
            client.publish(topic, message);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
        System.out.println("消息发布");
        System.out.println("broker: " + broker);
        System.out.println("clientId: " + client.getClientId());
        System.out.println("topic: " + topic);
        System.out.println("content: " + content);
    }

    @GetMapping("/subscribe")
    public void subscribe(@RequestParam(value = "broker") String broker,
                          @RequestParam(value = "topic") String topic){
        MqttClient client = mqttClientMap.get(broker);
        if(ObjectUtils.isEmpty(client)){
            System.out.println("mqtt服务器尚未连接");
        }

        Integer status = clientStatusMap.get(broker);
        if(SLEEPING.equals(status)){
            try {
                System.out.println("重新唤醒mqtt服务器" + broker);
                client.connect();
                System.out.println("成功唤醒mqtt服务器" + broker);
            } catch (MqttException e) {
                throw new RuntimeException(e);
            }
            clientStatusMap.put(broker, RUNNING);
        }

        try {
            client.subscribe(topic);
            System.out.println("消息订阅");
            System.out.println("broker: " + broker);
            System.out.println("clientId: " + client.getClientId());
            System.out.println("topic: " + topic);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }

        // 设置回调
        client.setCallback(new MqttCallback() {

            public void connectionLost(Throwable cause) {
                System.out.println("connectionLost: " + cause.getMessage());
            }

            public void messageArrived(String topic, MqttMessage message) {
                System.out.println("订阅消息已到达---------");
                System.out.println("topic: " + topic);
                System.out.println("Qos: " + message.getQos());
                System.out.println("message content: " + new String(message.getPayload()));

                // todo 获取到payload，应该根据消息的内容进行处理
            }

            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("deliveryComplete---------" + token.isComplete());
            }

        });
    }

    @GetMapping("/disconnect")
    public void disconnect(@RequestParam(value = "broker") String broker){
        MqttClient client = mqttClientMap.get(broker);
        if(ObjectUtils.isEmpty(client)){
            System.out.println("mqtt服务器尚未连接");
        }

        try {
            // disconnection() 只是客户端停止监听，重新使用只需要connect()
            clientStatusMap.put(broker, SLEEPING);
            client.disconnect();
            System.out.println("mqtt客户端停止监听服务器消息");
            System.out.println("broker: " + broker);
            System.out.println("clientId: " + client.getClientId());
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    @GetMapping("/close")
    public void close(@RequestParam(value = "broker") String broker){
        MqttClient client = mqttClientMap.get(broker);
        if(ObjectUtils.isEmpty(client)){
            System.out.println("mqtt服务器尚未连接");
        }
        mqttClientMap.remove(broker);
        clientStatusMap.remove(broker);

        try {
            client.disconnect();
            client.close();
            System.out.println("mqtt客户端释放所有资源");
            System.out.println("broker: " + broker);
            System.out.println("clientId: " + client.getClientId());
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }
}
